package com.surfilter.massdata.spark.task.etl;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.regex.Pattern;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.HiveContext;
import org.apache.spark.storage.StorageLevel;

import scala.Tuple2;

import com.act.sparkanalyz.log.SysLog;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.surfilter.massdata.spark.bean.DnsRaw;
import com.surfilter.massdata.spark.bean.Dns_tmp;
import com.surfilter.massdata.spark.bean.Domain;
import com.surfilter.massdata.spark.bean.ExDomain;
import com.surfilter.massdata.spark.bean.ExDomainRegex;
import com.surfilter.massdata.spark.util.BaseRedisDao;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.ConfigUtil;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.StringUtil;
import com.surfilter.massdata.spark.util.URLUtil;

public class EtlDayTask {
	public static String path = ConfigUtil.get("dnsfile_path", "");
	public static String[] DIps = ConfigUtil.getArray("surfilter.114.ips");
	public static String[] CIps = ConfigUtil.getArray("surfilter.filter.cips");
	public static String filterDIps = "";
	public static String filterCIps = "";
	public static String URL_PREFIX_HTTP = "http://";
	public static String DOMIAIN_REGX = "^([a-zA-Z0-9]([a-zA-Z0-9\\-]{0,61}[a-zA-Z0-9])?\\.)+[a-zA-Z]{2,6}$";
	public static Pattern domainPattern = Pattern.compile(DOMIAIN_REGX);

	public static void main(String[] args) {

		String dnsfile_path = "";
		String dayStr = "";
		Date date = null;
		String dns_src_hivedb = ConfigUtil.get("dns_src_hivedb", "default");
		String dns_src_table = ConfigUtil.get("dns_src_table", "dns");
		String hiveDB = ConfigUtil.get("hiveDB", "myhive");
		String dns_etl_table = ConfigUtil.get("dns_etl_table",
				"d_dns_domain_hour");
		String domain_etl_table = ConfigUtil.get("domain_etl_table",
				"day_domain");
		try {
			for (String ip : DIps) {
				filterDIps = filterDIps + ip + ",";
			}
			for (String ip : CIps) {
				filterCIps = filterCIps + ip + ",";
			}
			System.out.println("filterDIps:"+filterDIps);
			System.out.println("filterCIps:"+filterCIps);
			SparkConf conf = new SparkConf().setAppName("etl_task");
			JavaSparkContext context = new JavaSparkContext(conf);
			HiveContext hiveContext = new HiveContext(context);
		if (args.length > 0) {
				// 时间格式:2017041230 %Y%m%d%H
				dayStr = args[0];
			} else {
				System.out.println("请输入时间参数，格式：2017041230");
				System.exit(1);
			}
			date = DateUtil.getExecDate(DateUtil.getYesterday(), dayStr);
			String time = DateUtil.getCurrentymd(date);//2017-04-12
			String day = DateUtil.getyyyyMMddHH(date);//20170412
			String hour=DateUtil.getCurrentHour_STR(date);//HH
			int hour_i=DateUtil.getCurrentHour(date);
			
			Function<Row, Boolean> f;
			String sql="select host,destip,srcip,count,accesstime  from "
					+ dns_src_hivedb + "." + dns_src_table
					+ " where day=" + day + " and hour ="+hour_i;
			System.out.println("sql:"+sql);
	           
			JavaRDD<Row> rdd = hiveContext
			.sql(sql).toJavaRDD()
			.filter(new dnsfilter(filterDIps,filterCIps,domainPattern))/*.persist(StorageLevel.MEMORY_AND_DISK_SER())*/ ;
	
			
            System.out.println("rdd.getNumPartitions:"+rdd.getNumPartitions());
            rdd.coalesce(500);
            System.out.println("rdd.getNumPartitions:"+rdd.getNumPartitions());
			JavaPairRDD<String, String> exd1 = rdd
					.mapToPair(new PairFunction<Row, String, Integer>() {
						private static final long serialVersionUID = 1L;

						@Override
						public Tuple2<String, Integer> call(Row t)
								throws Exception {
							String domain = t.getAs("host").toString() ;
							String dip = t.getAs("destip").toString() ;
							return new Tuple2<String, Integer>(domain + "\t"
									+ dip, 1);
						}

					})
					.reduceByKey(new Function2<Integer, Integer, Integer>() {
						private static final long serialVersionUID = 1L;

						// domain IP v1
						@Override
						public Integer call(Integer v1, Integer v2)
								throws Exception {
							// TODO Auto-generated method stub
							return 1;
						}

					})
					.mapToPair(
							new PairFunction<Tuple2<String, Integer>, String, String>() {

								@Override
								public Tuple2<String, String> call(
										Tuple2<String, Integer> t)
										throws Exception {
									// key:domain dip
									String[] key = StringUtils
											.split(t._1, "\t");
									String topDomain = URLUtil
											.getDomainName(URL_PREFIX_HTTP
													+ key[0]);
									topDomain = topDomain.toLowerCase();
									return new Tuple2<String, String>(key[1]
											+ "\t" + topDomain, "1");
								}

							}).persist(StorageLevel.MEMORY_AND_DISK_SER());

			genExdomain_new(exd1);
			//exd1.unpersist();
			
			BaseRedisDao redisDao = new BaseRedisDao();
			Map<String, ExDomainRegex> exDomainMap = redisDao
					.getAllExDomainMap("EXDOMAIN");
			final Broadcast<Map<String, ExDomainRegex>> broadBswdMap = context
					.broadcast(exDomainMap);

			JavaRDD<Dns_tmp> dnsRDD = rdd
					.mapToPair(
							new PairIP_new(broadBswdMap.getValue(),
									domainPattern))
					.reduceByKey(
							new org.apache.spark.api.java.function.Function2<String, String, String>() {
								public String call(String v1, String v2)
										throws Exception {
									String str1[] = StringUtils.split(v1, "\t");
									String str2[] = StringUtils.split(v2, "\t");
									int num1 = Integer.valueOf(str1[0]);
									int num2 = Integer.valueOf(str2[0]);
									int num = num1 + num2;
									long time1 = Long.valueOf(str1[1]);
									long time2 = Long.valueOf(str2[1]);
									// 取最早发现时间
									long time = (time1 < time2 ? time1 : time2);
									String value = num + "\t" + time;
									return value;
								}
							}).mapPartitions(new flatMapDNS_new())
			 /*.persist(StorageLevel.MEMORY_AND_DISK_SER())*/ ;
			//rdd.unpersist();
			DataFrame dnsDF = hiveContext
					.createDataFrame(dnsRDD, Dns_tmp.class);
			dnsDF.registerTempTable("dns_t_t_t");
			String sql1 = "insert into table "
					+ hiveDB
					+ "."
					+ dns_etl_table
					+ "  partition( ds='"
					+ time
					+ "',hour="+hour+") select '',domain,cip,dip,dns_visit_count from dns_t_t_t";

			hiveContext.sql(sql1);
			hiveContext.dropTempTable("dns_t_t_t");

			// DIP省内IP过滤
			
			
			final TreeMap<Long, Long> ipMap = CommonUtils.getIpExtensionMap();
			final Broadcast<TreeMap<Long, Long>> broadBswdipMap = context.broadcast(ipMap);
			JavaRDD<Domain> domainRDD = dnsRDD
					.filter(new filterProvinceDIP_new(broadBswdipMap.getValue()))
					.mapToPair(new PairFunction<Dns_tmp, String, String>() {

						@Override
						public Tuple2<String, String> call(Dns_tmp line)
								throws Exception {
							String domain = line.getDomain();
							String dip = line.getDip();
							String visit_count = line.getDns_visit_count() + "";
							String findtime = line.getFindtime() + "";
							String key = domain + "\t" + dip;
							String value = visit_count + "\t" + findtime;
							return new Tuple2(key, value);
						}
					})
					.reduceByKey(
							new org.apache.spark.api.java.function.Function2<String, String, String>() {

								@Override
								public String call(String v1, String v2)
										throws Exception {
									String str1[] = StringUtils.split(v1, "\t");
									String str2[] = StringUtils.split(v2, "\t");
									int num1 = Integer.valueOf(str1[0]);
									int num2 = Integer.valueOf(str2[0]);
									int num = num1 + num2;
									long time1 = Long.valueOf(str1[1]);
									long time2 = Long.valueOf(str2[1]);
									// 取最早发现时间
									long time = (time1 < time2 ? time1 : time2);
									String value = num + "\t" + time;
									return value;
								}
							})
					.map(new Function<Tuple2<String, String>, Tuple2<String, String>>() {
						@Override
						public Tuple2<String, String> call(
								Tuple2<String, String> v1) throws Exception {
							String[] key = StringUtils.split(v1._1(), "\t");
							String domain = key[0];
							String dip = key[1];
							String value = v1._2();
							return new Tuple2<String, String>(domain, dip + "\t" + value);
						}
					})
					.mapToPair(
							new PairFunction<Tuple2<String, String>, String, String>() {

								@Override
								public Tuple2<String, String> call(
										Tuple2<String, String> line)
										throws Exception {
									return new Tuple2<String, String>(
											line._1(), line._2());
								}

							})
					.reduceByKey(new Function2<String, String, String>() {

						@Override
						public String call(String v1, String v2)
								throws Exception {
							// :DIP1 VISITSCOUNT1 FINDTIME
							String[] value_1 = StringUtils.split(v1, "\t");
							String[] value_2 = StringUtils.split(v2, "\t");
							long time1 = Long.valueOf(value_1[2]);
							long time2 = Long.valueOf(value_2[2]);
							// 取最早发现时间
							long time = (time1 < time2 ? time1 : time2);
							String value = value_1[0] + "|" + value_2[0] + "\t"
									+ value_1[1] + "|" + value_2[1] + "\t"
									+ time;
							return value;
						}

					}).mapPartitions(new flatMapDomain_new());

			DataFrame domainDF = hiveContext.createDataFrame(domainRDD,	Domain.class);
			domainDF.registerTempTable("domain_t_t_t");

			String sql2 = "insert into table "
					+ hiveDB
					+ "."
					+ domain_etl_table
					+ "  partition( ds='"
					+ time
					+ "',hour="+hour+") select domain,dip,findtime,dns_visit_count from domain_t_t_t";
			hiveContext.sql(sql2);
			hiveContext.dropTempTable("domain_t_t_t");
			dnsRDD.unpersist();
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private static void genExdomain_new(JavaPairRDD<String, String> exd1) {

		try {
			SysLog.info("*****************genExdomain************************");

			Map<String, Object> exd_count = exd1.countByKey();
			 System.out.println("exd_count.size:"+exd_count.size());
			Iterator it = exd_count.keySet().iterator();
			int judgeNum = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.judge.num", "200"));
			System.out.println("judgeNum:" + judgeNum);
			Map<String, Object> exdmap = new HashMap<String, Object>();
			// 过滤掉小于阀值的key
			while (it.hasNext()) {
				String key = it.next().toString();
				int num = Integer.valueOf(exd_count.get(key).toString());
				if (num > judgeNum) {
					exdmap.put(key, 1);
				}
			}

			int maxLimit = Integer.valueOf(ConfigUtil.get(
					"surfilter.massdata.exdomain.max.limit", "10000"));
			exd1.filter(new exdFilter_new(exdmap))
					.mapToPair(new transDomainKeyMapToPair_new())
					.reduceByKey(new ipMerge_new())
					.foreachPartition(new FlatMapExDomain_new(maxLimit));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

}

class exdFilter_new implements Function<Tuple2<String, String>, Boolean> {
	private static final long serialVersionUID = 1L;
	Map<String, Object> exd_count = null;

	public exdFilter_new(Map<String, Object> exd_count) {
		this.exd_count = exd_count;
	}

	@Override
	public Boolean call(Tuple2<String, String> v1) throws Exception {
		String key = v1._1();
		if (exd_count.containsKey(key)) {
			return true;
		} else {
			return false;
		}
	}

}

class FlatMapExDomain_new implements
		VoidFunction<Iterator<Tuple2<String, String>>> {
	private static final long serialVersionUID = 1L;
	int maxLimit = 10000;

	FlatMapExDomain_new(int maxLimit) {
		this.maxLimit = maxLimit;
	}

	@Override
	public void call(Iterator<Tuple2<String, String>> t) throws Exception {
		String EX_DOMAIN_REDIS_KEY = "EXDOMAIN";
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		BaseRedisDao redisDao = new BaseRedisDao();
		ObjectMapper mapper = new ObjectMapper();
		ObjectWriter writer = mapper.writer();

		while (t.hasNext()) {
			// key:<domaim> value:<dip1,dip2>
			Tuple2<String, String> tt = t.next();
			String domain = tt._1();
			String value = tt._2();

			ExDomain exDomain = new ExDomain();
			// 当前数据找到的域名、ip
			String[] value_list = StringUtils.split(value, ",");
			// redis 里存的域名、ip
			String redis_value = redisDao.hgetExDomain(EX_DOMAIN_REDIS_KEY,
					domain);
			Set ipSet = new HashSet();
			String ips_new = "";
			if (redis_value != null) {
				ExDomain redis_domain = mapper.readValue(
						StringUtil.toLowerJson(redis_value), ExDomain.class);
				String redis_ips = redis_domain.getIps();
				String[] ip_list = StringUtils.split(redis_ips, ",");
				for (String i : ip_list) {
					ipSet.add(i.trim());
				}
			}
			if (ipSet.contains(EX_DOMAIN_DEFAULT_IP)) {

			} else {
				for (String i : value_list) {
					ipSet.add(i.trim());
				}

				if (ipSet.size() > maxLimit) {
					ips_new = EX_DOMAIN_DEFAULT_IP;
				} else {
					Iterator iterator = ipSet.iterator();
					while (iterator.hasNext()) {
						ips_new += iterator.next();
						ips_new += ",";
					}
					ips_new = ips_new.substring(0, ips_new.length() - 1);
				}

				exDomain.setDomain("www." + domain);
				exDomain.setIps(ips_new);
				exDomain.setSuffix(domain);
				exDomain.setFindtime(new SimpleDateFormat("yyyyMMdd")
						.format(new Date()));
				// System.out.println("redis:" +
				// StringUtil.toUpperJson(writer.writeValueAsString(exDomain)));
				redisDao.hsetExDomain(EX_DOMAIN_REDIS_KEY, domain, StringUtil
						.toUpperJson(writer.writeValueAsString(exDomain)));

			}
		}
	}
}

class transDomainKeyMapToPair_new implements
		PairFunction<Tuple2<String, String>, String, String> {

	@Override
	public Tuple2<String, String> call(Tuple2<String, String> t)
			throws Exception {
		String[] key = StringUtils.split(t._1(), "\t");
		String ip = key[0];
		String domain = key[1];
		return new Tuple2<String, String>(domain, ip);
	}

}

class ipMerge_new implements Function2<String, String, String> {

	@Override
	public String call(String v1, String v2) throws Exception {
		if (v1.contains(v2)) {
			return v1;
		} else if (v2.contains(v1)) {
			return v2;
		} else {
			return v1 + "," + v2;
		}

	}

}

class PairIP_new implements PairFunction<Row, String, String> {

	String URL_PREFIX_HTTP = "http://";
	Pattern domainPattern;
	int flag = 0;
	Map<String, ExDomainRegex> exDomainMap;

	public PairIP_new(Map<String, ExDomainRegex> exDomainMap,
			Pattern domainPattern) {
		this.exDomainMap = exDomainMap;
		this.domainPattern = domainPattern;
	}

	@Override
	public Tuple2<String, String> call(Row t) throws Exception {
		String domain = t.getAs("host").toString();
		String dip = t.getAs("destip").toString();
		String cip = t.getAs("srcip").toString();
		String dns_visit_count = t.getAs("count").toString();
		String findtime = t.getAs("accesstime").toString();
		// guoji.news.baidu.com
		String topDomain = URLUtil.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return null;
		}
		topDomain = topDomain.toLowerCase();
		String exDomain = domain;
		// 泛域名处理
		try {
			// 缺陷【smcs_all_v1.0.0.0】
			domain = exdomainDeal(topDomain, dip, exDomainMap);
		} catch (Exception e) {
			domain = null;
		}

		if (domain == null) {
			domain = exDomain;
			exDomain = null;
		}
		String key = "";
		key = domain + "\t" + dip + "\t" + cip;
		String value = "";
		if (this.flag == 0) {
			// findtime 转成时间戳 2017-05-02 06:00:00
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");// 小写的mm表示的是分钟
			java.util.Date date = sdf.parse(findtime);
			long time = date.getTime() / 1000;
			value = dns_visit_count + "\t" + time;
		} else {
			value = dns_visit_count + "\t" + findtime;
		}

		return new Tuple2(key, value);
	}

	/**
	 * 过滤掉泛域名-IP,将匹配到的泛域名-IP替换成对应的域名-IP
	 * 
	 * @param domain
	 * @param dips
	 * @return
	 */

	private String exdomainDeal(String topDomain, String dips,
			Map<String, ExDomainRegex> exDomainMap) {
		String EX_DOMAIN_DEFAULT_IP = "0.0.0.0";
		if (StringUtils.isBlank(topDomain) || StringUtils.isBlank(dips)) {
			return null;
		}
		List<String> ipList = new ArrayList<String>();
		ipList.addAll(Arrays.asList(StringUtils.split(dips, "\\|")));
		String exDomain = null;
		if (exDomainMap.containsKey(topDomain)) {
			ExDomainRegex exDomainRegex = exDomainMap.get(topDomain);
			Set<String> exIps = exDomainRegex.getIpSet();
			if (exIps.contains(EX_DOMAIN_DEFAULT_IP)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			} else if (intersectionSet(ipList, exIps)) {
				exDomain = exDomainRegex.getNewName();
				return exDomain;
			}
		}
		return exDomain;
	}

	private boolean intersectionSet(List<String> list, Set<String> setB) {
		list.retainAll(setB);
		return list.size() > 0;
	}
}

class flatMapDNS_new implements
		FlatMapFunction<Iterator<Tuple2<String, String>>, Dns_tmp> {

	@Override
	public Iterable<Dns_tmp> call(Iterator<Tuple2<String, String>> t)
			throws Exception {
		List<Dns_tmp> list = new ArrayList<Dns_tmp>();
		// map key:domain + "\t" + dip + "\t" + cip
		// value: 1+"\t"+ 时间戳
		while (t.hasNext()) {
			Tuple2<String, String> tt = t.next();
			String[] key = StringUtils.split(tt._1(), "\t");
			String[] value = StringUtils.split(tt._2(), "\t");
			long findtime = Long.valueOf(value[1]);
			Date date = new Date(findtime * 1000);
			int hour = DateUtil.getCurrentHour(date);
			String hourStr = "";
			if (hour < 10) {
				hourStr = "0" + hour;
			} else {
				hourStr = String.valueOf(hour);
			}
			Dns_tmp e = new Dns_tmp();
			e.setDomain(key[0]);
			e.setDip(key[1]);
			e.setCip(key[2]);
			e.setDns_visit_count(Long.valueOf(value[0]));
			e.setFindtime(findtime);
			e.setHour(hourStr);
			list.add(e);
		}
		return list;
	}
}

class filterProvinceDIP_new implements Function<Dns_tmp, Boolean> {

	final TreeMap<Long, Long> ipMap ;
	filterProvinceDIP_new(TreeMap<Long, Long> ipMap){
		this.ipMap=ipMap;
	}
	@Override
	public Boolean call(Dns_tmp v1) throws Exception {
		 String dip = v1.getDip();
		 boolean flag = CommonUtils.isExists(ipMap, dip);
		 return flag;
		//return true;
	}
}

class flatMapDomain_new implements
		FlatMapFunction<Iterator<Tuple2<String, String>>, Domain> {

	@Override
	public Iterable<Domain> call(Iterator<Tuple2<String, String>> t)
			throws Exception {

		List<Domain> list = new ArrayList<Domain>();
		// <domaim : dip1|dip2 dns_visit_count1|dns_visit_count2 findtime>
		while (t.hasNext()) {
			Tuple2<String, String> tt = t.next();
			String key = tt._1();
			String[] value = StringUtils.split(tt._2(), "\t");
			long findtime = Long.valueOf(value[2]);
			Date date = new Date(findtime * 1000);
			int hour = DateUtil.getCurrentHour(date);
			String hourStr = "";
			if (hour < 10) {
				hourStr = "0" + hour;
			} else {
				hourStr = String.valueOf(hour);
			}
			Domain e = new Domain();
			e.setDomain(key);
			e.setDip(value[0]);
			e.setDns_visit_count(value[1]);
			e.setFindtime(findtime);
			e.setHour(hourStr);
			list.add(e);
		}
		return list;

	}

}


class dnsfilter implements Function<Row, Boolean> {
	String filterDIps;
	String filterCIps;
	String URL_PREFIX_HTTP = "http://";
	Pattern domainPattern ;
	private static final long serialVersionUID = 1L;
    dnsfilter(String filterDIps,String filterCIps,Pattern domainPattern){
    	this.filterDIps=filterDIps;
    	this.filterCIps=filterCIps;
    	this.domainPattern=domainPattern;
    	
    }
	@Override
	public Boolean call(Row t) throws Exception {
		String domain = t.getAs("host").toString() ;
		String dip = t.getAs("destip").toString() ;
		String cip = t.getAs("srcip").toString() ;

		// 目的IP 114IP过滤
		if (filterDIps.contains(dip)) {
			return false;
		}
		// 源IP CIP过滤
		if (filterCIps.contains(cip)) {
			return false;
		}
		if (StringUtils.isBlank(domain)
				|| StringUtils.isBlank(dip)
				|| StringUtils.isBlank(cip)) {
			return false;
		}
		// 域名合法性校验
		if (!domainPattern.matcher(domain).matches()) {
			return false;
		}

		String topDomain = URLUtil
				.getDomainName(URL_PREFIX_HTTP + domain);
		if (topDomain == null) {
			return false;
		}
		return true;
	}

}
